package com.unionpay.sqoop

import java.util
import java.util.UUID

import org.apache.sqoop.client.SqoopClient
import org.apache.sqoop.model.{MJob, MLink}

object Sql2Hdfs {


  def main(args: Array[String]): Unit = {

    val url = "http://localhost:12000/sqoop/"
    val client = new SqoopClient(url)

    //    client.deleteJob("mysql2hdfs")
    //    client.deleteLink("JDBC")
    //    client.deleteLink("HDFS")


    val fromLink = getMySqlLink(client, "jdbc-link-" + UUID.randomUUID().toString().substring(0, 20))
    client.saveLink(fromLink)
    println(fromLink.getConnectorName)
    println(fromLink.getName)


    val destLink = getHDFSLink(client, "hdfs-link-" + UUID.randomUUID().toString().substring(0, 20))
    client.saveLink(destLink)
    println(destLink.getConnectorName)
    println(destLink.getName)

    //    val job = getHdfs2MySQLJob(client, destLink, fromLink, "hdfs2mysql" + UUID.randomUUID().toString().substring(0, 20))
    //    client.saveJob(job)

    val job = getMysql2HdfSJob(client, fromLink, destLink, "mysql2hdfs-" + UUID.randomUUID().toString().substring(0, 20))
    client.saveJob(job)

    println(job.getName)
    startJob(client, job)
  }

  /**
    *
    * @param client
    * @param linkName
    * @return
    */
  def getMySqlLink(client: SqoopClient, linkName: String): MLink = {

    val mysqlConnectorName: String = "generic-jdbc-connector"

    val link = client.createLink(mysqlConnectorName)

    link.setName(linkName)

    val linkConfig = link.getConnectorLinkConfig
    linkConfig.getStringInput("linkConfig.connectionString").setValue("jdbc:mysql://localhost:3306/hive")
    linkConfig.getStringInput("linkConfig.jdbcDriver").setValue("com.mysql.jdbc.Driver")
    linkConfig.getStringInput("linkConfig.username").setValue("root")
    linkConfig.getStringInput("linkConfig.password").setValue("root")
    linkConfig.getStringInput("dialect.identifierEnclose").setValue(" ")


    println("***** mysql link information *****" + link)


    link
  }

  /**
    *
    * @param client
    * @param linkName
    * @return
    */

  def getHDFSLink(client: SqoopClient, linkName: String): MLink = {

    val hdfsConnectorName: String = "hdfs-connector"
    val link = client.createLink(hdfsConnectorName)

    link.setName(linkName)

    val linkConfig = link.getConnectorLinkConfig
    linkConfig.getStringInput("linkConfig.uri").setValue("hdfs://localhost:9000")
    linkConfig.getStringInput("linkConfig.confDir").setValue("/usr/local/software/hadoop-2.7.6/etc/hadoop/")

    println("***** hdfs link information *****" + link)

    link
  }

  /**
    *
    * @param client
    * @param fromLink
    * @param toLink
    * @param jobName
    * @return
    */

  def getHdfs2MySQLJob(client: SqoopClient, fromLink: MLink, toLink: MLink, jobName: String): MJob = {

    val job = client.createJob(fromLink.getName, toLink.getName)

    job.setName(jobName)

    val fromJobConfig = job.getFromJobConfig

    fromJobConfig.getStringInput("fromJobConfig.inputDirectory").setValue("/user/lspie/hive/warehouse/sqoop/input/")


    val toJobConfig = job.getToJobConfig
    toJobConfig.getStringInput("toJobConfig.schemaName").setValue("hive")
    toJobConfig.getStringInput("toJobConfig.tableName").setValue("my_user")
    //    toJobConfig.getListInput("toJobConfig.columnList").setValue(util.Arrays.asList(
    //      "uid", "movieid", "rating", "uninxtime"))

    job.getDriverConfig.getIntegerInput("throttlingConfig.numExtractors").setValue(1)

    println("***** toJobConfig *****" + toJobConfig)

    job
  }

  def getMysql2HdfSJob(client: SqoopClient, fromLink: MLink, toLink: MLink, jobName: String): MJob = {
    val job = client.createJob(fromLink.getName, toLink.getName)

    job.setName(jobName)

    val fromJobConfig = job.getFromJobConfig

    fromJobConfig.getStringInput("fromJobConfig.schemaName").setValue("hive")
    fromJobConfig.getStringInput("fromJobConfig.tableName").setValue("my_user")
    fromJobConfig.getListInput("fromJobConfig.columnList").setValue(util.Arrays.asList(
      "id", "account", "passwd"))
    fromJobConfig.getStringInput("fromJobConfig.partitionColumn").setValue("id")

    val toJobConfig = job.getToJobConfig

    toJobConfig.getStringInput("toJobConfig.outputDirectory").setValue("/user/lspie/hive/warehouse/sqoop/output/")
    toJobConfig.getEnumInput("toJobConfig.outputFormat").setValue("TEXT_FILE")
    toJobConfig.getEnumInput("toJobConfig.compression").setValue("NONE")
    toJobConfig.getBooleanInput("toJobConfig.overrideNullValue").setValue(true)

    job.getDriverConfig.getIntegerInput("throttlingConfig.numExtractors").setValue(1)

    job

  }

  /**
    *
    * @param client
    * @param job
    */

  def startJob(client: SqoopClient, job: MJob) = {

    val submission = client.startJob(job.getName)

    while (submission.getStatus.isRunning && submission.getProgress != -1)
      println("process:" + submission.getProgress * 100)
  }

}
